package rete;


import alink.AlsRateRecommStreamOpTest;
import com.asap.demo.function.dealMapFunction;
import com.asap.demo.function.dealStreamProcessFunctionList;
import com.asap.demo.model.BeanField;
import com.asap.demo.sourcefunc.MysqlSourceFunction1;
import com.asap.demo.table.JsonField;
import com.asap.demo.table.RuleParse;
import com.asap.demo.utils.Constants;
import com.asap.demo.utils.DateUtil;
import com.asap.demo.utils.Utils;
import com.asap.interf.Action;
import com.asap.rule.StandardEvent;
import com.asap.rule.engine.InferenceEngine;
import com.asap.rule.engine.PatternMatcher;
import com.asap.rule.orm.DbFetcher;
import com.asap.rule.util.PropTransformMap;
import com.asap.rule.util.RuleReader;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.typeutils.ListTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.test.util.MiniClusterWithClientResource;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
import org.junit.ClassRule;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import java.io.IOException;
import java.sql.Timestamp;
import java.util.*;

public class ReteDemo4 {

	private static final Logger logger = LoggerFactory.getLogger(AlsRateRecommStreamOpTest.class);

	@ClassRule
	public static MiniClusterWithClientResource flinkCluster =
			new MiniClusterWithClientResource(
					new MiniClusterResourceConfiguration.Builder()
							.setNumberSlotsPerTaskManager(3)
							.setNumberTaskManagers(2)
							.build());
	@Test
	public void test() throws Exception {

		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);
		env.enableCheckpointing(5000);  //检查点 每5000ms
		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
		EnvironmentSettings blinkStreamSettings = EnvironmentSettings.newInstance()
				.useBlinkPlanner()
				.inStreamingMode()
				.build();
		StreamTableEnvironment blinkStreamTableEnv = StreamTableEnvironment.create(env, blinkStreamSettings);

		Properties browseProperties = new Properties();
		browseProperties.put("bootstrap.servers", "192.168.1.25:9093");
		browseProperties.put("group.id", "temporal");
		browseProperties.put("auto.offset.reset", "latest");
		PropTransformMap.getInstance().readConfigMap("./conf/cfg.properties");
		Map<String, String> configMap = new HashMap<String, String>();
		configMap.put(Constants.DB_JDBC_USER, "root");
		configMap.put(Constants.DB_JDBC_PASSWD, "1qazXSW@3edc");
		configMap.put(Constants.DB_JDBC_URL, "jdbc:mysql://10.28.184.25:3306/SSA?useUnicode=true&characterEncoding=utf-8");
		configMap.put(Constants.DB_JDBC_DRIVER, "com.mysql.jdbc.Driver");
		configMap.put(Constants.INITAL_POOL_SIZE, "10");
		configMap.put(Constants.MIN_POOL_SIZE, "5");
		configMap.put(Constants.MAX_IDLE_TIME, "50");
		configMap.put(Constants.MAX_STATE_ELEMENTS, "100");
		configMap.put(Constants.MAX_IDLE_TIME, "60");
		DbFetcher dbFetcher = new DbFetcher(configMap);
		List<String> listRule = RuleReader.readRules(dbFetcher);
		System.out.println("ListRule::" + listRule.size());

		final String RULE_SBROAD_CAST_STATE = "RulesBroadcastState";

		RuleParse ruleParse = new RuleParse();
		Map properties = new HashMap();
		ruleParse.parseData("./conf/cfg.json");

		//1、读取mysql的配置消息
		DataStream<List<String>> conf = env.addSource(new MysqlSourceFunction1(dbFetcher));

		//2、创建MapStateDescriptor规则，对广播的数据的数据类型的规则
		MapStateDescriptor<String, List<String>> ruleStateDescriptor = new MapStateDescriptor<>(RULE_SBROAD_CAST_STATE
				, BasicTypeInfo.STRING_TYPE_INFO
				, new ListTypeInfo<>(String.class));
		//3、对conf进行broadcast返回BroadcastStream
		final BroadcastStream<List<String>> confBroadcast = conf.broadcast(ruleStateDescriptor);

		//DataStream<String> dataStream = env.fromElements("{\"ORG_ID\":\"1\",\"RAW_MSG\":\"useradd,su - root\",\"EVENT_THREE_TYPE\":\"40001\",\"EVENT_TWO_TYPE\":\"40001\",\"SRC_PORT\":\"123\",\"DST_PORT\":\"124\",\"DST_IP\":\"10.16.254.11\",\"SRC_IP\":\"50.115.134.50\",\"CREATE_TIME\":\"2021-07-09 18:15:21.001\",\"DEVICE_PARENT_TYPE\":\"LINUX\",\"SNOW_ID\":\"85512\",\"EVENT_THREE_TYPE_DESC\":\"暴力破解失败\",\"ts\":\"2021-05-27 16:06:58\",\"ACCOUNT\":\"asap\",\"collectionName\":\"bwdOMS\",\"eRuleId\":\"0\",\"RULE_TJ_COUNT\":11,\"TAGS\":{\"EVENT_ONE_TYPE\":\"20000\",\"DIRECTION\":\"内部\",\"EVENT_TWO_TYPE\":\"10015\",\"EVENT_THREE_TYPE\":\"20101\"},\"DEVICE_TYPE\":\"OSM\",\"DIRECTION\":\"0\"}\n");
		DataStream<String> dataStream = env.fromElements("{\"DIRECTION\":\"0\",\"ATTACK_STAGE\":\"命令控制\",\"DEVICE_PARENT_TYPE\":\"IPS\",\"URL\":\"www.baidu.com\",\"SRC_PORT\":\"58513\",\"DST_PORT\":\"31177\",\"RISK_LEVEL\":\"99\",\"SRC_ASSET_TYPE\":\"4\",\"SRC_ASSET_SUB_TYPE\":\"412\",\"DST_ASSET_TYPE\":\"4\",\"DST_ASSET_SUB_TYPE\":\"412\",\"SRC_POST\":\"1\",\"DST_POST\":\"0\",\"INSERT_TIME\":\"2021-05-01 00:00:00.000\",\"DST_ASSET_NAME\":\"ddde\",\"SRC_ASSET_NAME\":\"wangwu\",\"SCENE_ID\":-5216633060008277343,\"SOURCE\":\"4\",\"ASSET_IP\":\"73.243.143.114\",\"TENANT_ID\":\"-1\",\"ORG_ID\":\"1\",\"DST_IP\":\"192.118.8.218\",\"EVENT_TYPE\":\"1008\",\"SRC_IP\":\"153.79.42.45\",\"CUSTOM_VALUE1\":\"187.36.226.184\",\"CUSTOM_VALUE2\":\"41.68.25.104\",\"SRC_PROVINCE\":\"日本\",\"SNOW_ID\":\"469260993\",\"DEVICE_TYPE\":\"TDA\",\"MESSAGE\":\"\",\"CHARACTER\":\"\",\"CUSTOM_LABEL1\":\"控制IP\",\"CREATE_TIME\":\"2021-04-26 17:04:18.000\",\"CUSTOM_LABEL2\":\"受控IP\",\"TYPE\":\"未知\",\"MALWARE_TYPE\":\"其他\",\"collectTime\":\"2021-04-29 19:40:36.000\",\"RULE_ID\":\"180607832\",\"DEVICE_IP\":\"239.150.69.203\",\"SRC_CITY\":\"日本\",\"recordTime\":\"2021-04-30 11:36:00.757\",\"equIP\":\"20.222.74.177\",\"DST_ZONE\":\"中国-湖北-武汉\",\"SERVERITY\":\"99\",\"SRC_COUNTRY\":\"日本\",\"SRC_ZONE\":\"意大利\",\"DST_COUNTRY\":\"中国\",\"DST_CITY\":\"广东\",\"DST_PROVINCE\":\"南京\",\"EVENT_ONE_TYPE\":\"-1\",\"EVENT_TWO_TYPE\":\"40001\",\"EVENT_NAME\":\"流氓软件\",\"EVENT_THREE_TYPE\":\"40001\",\"eRuleId\":\"0\",\"hitRuleIds\":\"1453556891900301314:Ordinary:A\"}",
				"{\"DIRECTION\":\"0\",\"ATTACK_STAGE\":\"命令控制\",\"DEVICE_PARENT_TYPE\":\"IPS\",\"URL\":\"www.baidu.com\",\"SRC_PORT\":\"58513\",\"DST_PORT\":\"31177\",\"RISK_LEVEL\":\"99\",\"SRC_ASSET_TYPE\":\"4\",\"SRC_ASSET_SUB_TYPE\":\"412\",\"DST_ASSET_TYPE\":\"4\",\"DST_ASSET_SUB_TYPE\":\"412\",\"SRC_POST\":\"1\",\"DST_POST\":\"0\",\"INSERT_TIME\":\"2021-05-01 00:00:00.000\",\"DST_ASSET_NAME\":\"ddde\",\"SRC_ASSET_NAME\":\"wangwu\",\"SCENE_ID\":-5216633060008277343,\"SOURCE\":\"4\",\"ASSET_IP\":\"73.243.143.114\",\"TENANT_ID\":\"-1\",\"ORG_ID\":\"1\",\"DST_IP\":\"192.118.8.218\",\"EVENT_TYPE\":\"1008\",\"SRC_IP\":\"153.79.42.45\",\"CUSTOM_VALUE1\":\"187.36.226.184\",\"CUSTOM_VALUE2\":\"41.68.25.104\",\"SRC_PROVINCE\":\"日本\",\"SNOW_ID\":\"469260998\",\"DEVICE_TYPE\":\"TDA\",\"MESSAGE\":\"\",\"CHARACTER\":\"\",\"CUSTOM_LABEL1\":\"控制IP\",\"CREATE_TIME\":\"2021-04-25 17:04:18.000\",\"CUSTOM_LABEL2\":\"受控IP\",\"TYPE\":\"未知\",\"MALWARE_TYPE\":\"其他\",\"collectTime\":\"2021-04-29 19:40:36.000\",\"RULE_ID\":\"180607832\",\"DEVICE_IP\":\"239.150.69.203\",\"SRC_CITY\":\"日本\",\"recordTime\":\"2021-04-30 11:36:00.757\",\"equIP\":\"20.222.74.177\",\"DST_ZONE\":\"中国-湖北-武汉\",\"SERVERITY\":\"99\",\"SRC_COUNTRY\":\"日本\",\"SRC_ZONE\":\"意大利\",\"DST_COUNTRY\":\"中国\",\"DST_CITY\":\"广东\",\"DST_PROVINCE\":\"南京\",\"EVENT_ONE_TYPE\":\"-1\",\"EVENT_TWO_TYPE\":\"40001\",\"EVENT_NAME\":\"流氓软件\",\"EVENT_THREE_TYPE\":\"40001\",\"eRuleId\":\"0\",\"hitRuleIds\":\"1453556891900301314:Ordinary:A\"}",
				"{\"DIRECTION\":\"0\",\"ATTACK_STAGE\":\"命令控制\",\"DEVICE_PARENT_TYPE\":\"IPS\",\"URL\":\"www.baidu.com\",\"SRC_PORT\":\"58513\",\"DST_PORT\":\"31177\",\"RISK_LEVEL\":\"99\",\"SRC_ASSET_TYPE\":\"4\",\"SRC_ASSET_SUB_TYPE\":\"412\",\"DST_ASSET_TYPE\":\"4\",\"DST_ASSET_SUB_TYPE\":\"412\",\"SRC_POST\":\"1\",\"DST_POST\":\"0\",\"INSERT_TIME\":\"2021-05-01 00:00:00.000\",\"DST_ASSET_NAME\":\"ddde\",\"SRC_ASSET_NAME\":\"wangwu\",\"SCENE_ID\":-5216633060008277343,\"SOURCE\":\"4\",\"ASSET_IP\":\"73.243.143.114\",\"TENANT_ID\":\"-1\",\"ORG_ID\":\"1\",\"DST_IP\":\"192.118.8.218\",\"EVENT_TYPE\":\"1008\",\"SRC_IP\":\"153.79.42.45\",\"CUSTOM_VALUE1\":\"187.36.226.184\",\"CUSTOM_VALUE2\":\"41.68.25.104\",\"SRC_PROVINCE\":\"日本\",\"SNOW_ID\":\"469260998\",\"DEVICE_TYPE\":\"TDA\",\"MESSAGE\":\"\",\"CHARACTER\":\"\",\"CUSTOM_LABEL1\":\"控制IP\",\"CREATE_TIME\":\"2021-04-25 17:04:38.000\",\"CUSTOM_LABEL2\":\"受控IP\",\"TYPE\":\"未知\",\"MALWARE_TYPE\":\"其他\",\"collectTime\":\"2021-04-29 19:40:36.000\",\"RULE_ID\":\"180607832\",\"DEVICE_IP\":\"239.150.69.203\",\"SRC_CITY\":\"日本\",\"recordTime\":\"2021-04-30 11:36:00.757\",\"equIP\":\"20.222.74.177\",\"DST_ZONE\":\"中国-湖北-武汉\",\"SERVERITY\":\"99\",\"SRC_COUNTRY\":\"日本\",\"SRC_ZONE\":\"意大利\",\"DST_COUNTRY\":\"中国\",\"DST_CITY\":\"广东\",\"DST_PROVINCE\":\"南京\",\"EVENT_ONE_TYPE\":\"-1\",\"EVENT_TWO_TYPE\":\"40001\",\"EVENT_NAME\":\"流氓软件\",\"EVENT_THREE_TYPE\":\"40001\",\"eRuleId\":\"0\",\"hitRuleIds\":\"1453556891900301314:Ordinary:A\"}");

		//DataStream<String> dataStream = env.fromElements("{\"ORG_ID\":\"1\",\"RAW_MSG\":\"useradd,su - root\",\"EVENT_THREE_TYPE\":\"40001\",\"EVENT_TWO_TYPE\":\"40001\",\"SRC_PORT\":\"123\",\"DST_PORT\":\"124\",\"DST_IP\":\"10.16.254.11\",\"SRC_IP\":\"50.115.134.50\",\"CREATE_TIME\":\"2021-07-09 18:15:21.001\",\"DEVICE_PARENT_TYPE\":\"LINUX\",\"SNOW_ID\":\"85512\",\"EVENT_THREE_TYPE_DESC\":\"暴力破解失败\",\"ts\":\"2021-05-27 16:06:58\",\"ACCOUNT\":\"asap\",\"collectionName\":\"bwdOMS\",\"eRuleId\":\"0\",\"RULE_TJ_COUNT\":11,\"TAGS\":{\"EVENT_ONE_TYPE\":\"20000\",\"DIRECTION\":\"内部\",\"EVENT_TWO_TYPE\":\"10015\",\"EVENT_THREE_TYPE\":\"20101\"},\"DEVICE_TYPE\":\"OSM\",\"DIRECTION\":\"0\"}");
		dataStream.print();
		DataStream<StandardEvent> kafkaData = getStandardEventDataStream(confBroadcast, dataStream);
		System.out.println("==============");
		kafkaData.print();
		DataStream<BeanField> kafkaData1 = kafkaData.map(new dealMapFunction(ruleParse)).setParallelism(1);
		Table inputTable = getTable(blinkStreamTableEnv, kafkaData1);
		blinkStreamTableEnv.createTemporaryView("InputTable", inputTable);

		String query = "select count(1) from InputTable";
		Table query1 = blinkStreamTableEnv.sqlQuery(query);
		blinkStreamTableEnv.toRetractStream(query1, Row.class).print("count(1)==");

		String querySQL0 = "select deviceType,count(1) from InputTable where deviceType='TDA' group by TUMBLE(createTime, INTERVAL '5' MINUTE),deviceType";
		String querySQL1 = "select * from (select deviceType,count(1) from InputTable where deviceType='TDA' group by TUMBLE(createTime, INTERVAL '5' MINUTE),deviceType) a,(select deviceType,count(1) from InputTable where deviceType='TDA' group by TUMBLE(createTime, INTERVAL '5' MINUTE),deviceType) b";


		String querySQL2 = "select deviceType,count(1) from InputTable where deviceType='OSM' group by TUMBLE(createTime, INTERVAL '5' MINUTE),deviceType ";

		String querySQL3 = "select window_start, window_end,deviceType,snowId,count(1) as cnt from TABLE(\n" +
				"    TUMBLE(TABLE InputTable, DESCRIPTOR(createTime), INTERVAL '10' MINUTES))" +
				" where deviceType='TDA' GROUP BY window_start, window_end,deviceType,snowId ";
		Table table2 = blinkStreamTableEnv.sqlQuery(querySQL3);
		blinkStreamTableEnv.toRetractStream(table2, Row.class).print("query1==");

//		String querySQL4 = "select b.* from (select deviceType,snowId ,count(1) as cnt from TABLE(\n" +
//				"TUMBLE(TABLE InputTable, DESCRIPTOR(createTime), INTERVAL '10' MINUTES))" +
//				" where deviceType='TDA' GROUP BY window_start, window_end,deviceType,snowId having count(1)=2) a,(SELECT * FROM InputTable) b" +
//				" where a.snowId =b.snowId ";
//		System.out.println(querySQL4);
//
//		table2 = blinkStreamTableEnv.sqlQuery(querySQL4);
//		blinkStreamTableEnv.toRetractStream(table2, BeanField.class).print("query2==");

//		String querySQL5 = "select deviceType,snowId from TABLE(\n" +
//				"TUMBLE(TABLE InputTable, DESCRIPTOR(createTime), INTERVAL '10' MINUTES))" +
//				" where deviceType='TDA' GROUP BY window_start, window_end,deviceType,snowId having count(1)>1";
//		Table resultTableIps1 = blinkStreamTableEnv.sqlQuery(querySQL5);
//		blinkStreamTableEnv.createTemporaryView("resultTableIps1", resultTableIps1);
//		String querySQL6 = "select InputTable.* from InputTable INNER JOIN resultTableIps1 ON InputTable.snowId = resultTableIps1.snowId";
//		Table resultTableIps = blinkStreamTableEnv.sqlQuery(querySQL6);
//		blinkStreamTableEnv.toRetractStream(resultTableIps, BeanField.class).print("BeanField==");

//
//		String sql = "select * from (select deviceType,count(1) from InputTable where deviceType='TDA' group by TUMBLE(createTime, INTERVAL '5' MINUTE),deviceType) a,(select deviceType,count(1) from InputTable where deviceType='IDS' group by TUMBLE(createTime, INTERVAL '5' MINUTE),deviceType) b,(select deviceType,count(1) from InputTable where deviceType='IDS' group by TUMBLE(createTime, INTERVAL '5' MINUTE),deviceType) c";
//		Table resultTableIps= blinkStreamTableEnv.sqlQuery(sql);
//		blinkStreamTableEnv.toRetractStream(resultTableIps, Row.class).print("queryIps==");
		env.execute("Broadcast test kafka");
	}

	private static DataStream<StandardEvent> getStandardEventDataStream(BroadcastStream<List<String>> confBroadcast, DataStream<String> dataStream) {

		DataStream<StandardEvent> dataStreamInfo = dataStream.map(new MapFunction<String, StandardEvent>() {
			@Override
			public StandardEvent map(String value) throws Exception {
				StandardEvent standardEvent = StandardEvent.parse(value);
				return standardEvent;
			}
		});


		return dataStreamInfo.assignTimestampsAndWatermarks(
				new AssignerWithPeriodicWatermarks<StandardEvent>() {
					Long currentMaxTimestamp = 0L;
					Long maxDelayTime = 5000L;

					@Override
					public long extractTimestamp(StandardEvent s, long l) {
						currentMaxTimestamp = Timestamp.valueOf(Utils.transforDate(s.getField("CREATE_TIME"))).getTime();
						return currentMaxTimestamp;
					}

					@Nullable
					@Override
					public Watermark getCurrentWatermark() {
						long time = currentMaxTimestamp - maxDelayTime;
						//logger.info("getCurrentWatermark.............."+time);
						return new Watermark(time);
					}
				}
		)
				.keyBy(new KeySelector<StandardEvent, String>() {
					@Override
					public String getKey(StandardEvent event) throws Exception {
						StringBuilder sb = new StringBuilder();
						sb.append(event.getField("DEVICE_PARENT_TYPE"))
								.append(event.getField("SNOW_ID"));
						return sb.toString();
					}
				})
				.connect(confBroadcast)
				.process(
						new dealStreamProcessFunctionList()
				).setParallelism(1);
	}

	private static Table getTable(StreamTableEnvironment blinkStreamTableEnv, DataStream<BeanField> kafkaData1) {
		return blinkStreamTableEnv.fromDataStream(kafkaData1,
				Schema.newBuilder()
						.column("snowId", DataTypes.STRING().notNull())
						.column("account","STRING")
						.column("action","INTEGER")
						.column("actionDesc","STRING")
						.column("assetIp","STRING")
						.column("attackStage","STRING")
						.column("ausIndex","STRING")
						.column("averageByteFlow","INTEGER")
						.column("averagePackageFlow","INTEGER")
						.column("baseline","STRING")
						.column("bizId","STRING")
						.column("bizName","STRING")
						.column("confidence","INTEGER")
						.column("createTime",DataTypes.TIMESTAMP_LTZ(3).notNull())
						.column("databaseName","STRING")
						.column("dataType","STRING")
						.column("deviceIp","STRING")
						.column("deviceName","STRING")
						.column("deviceParentType","STRING")
						.column("deviceType","STRING")
						.column("direction","STRING")
						.column("directionDesc","STRING")
						.column("downBaseLineFlow","INTEGER")
						.column("downFlowTotal","INTEGER")
						.column("dstAssetId","INTEGER")
						.column("dstAssetGroup","STRING")
						.column("dstAssetKey","INTEGER")
						.column("dstAssetName","STRING")
						.column("dstAssetPublic","INTEGER")
						.column("dstAssetStatus","INTEGER")
						.column("dstAssetSubType","STRING")
						.column("dstAssetType","INTEGER")
						.column("dstBizId","STRING")
						.column("dstCity","STRING")
						.column("dstCountry","STRING")
						.column("dstDomainName","STRING")
						.column("dstIntelDesc","STRING")
						.column("dstIntelId","STRING")
						.column("dstIntelType","INTEGER")
						.column("dstIp","STRING")
						.column("dstLatitude","STRING")
						.column("dstLongitude","STRING")
						.column("dstOrgId","STRING")
						.column("dstOrgName","STRING")
						.column("dstPort","STRING")
						.column("dstPost","STRING")
						.column("dstProvince","STRING")
						.column("dstSubDomainName","STRING")
						.column("eventName","STRING")
						.column("eventOneType","INTEGER")
						.column("eventThreeType","INTEGER")
						.column("eventTwoType","INTEGER")
						.column("eventThreeTypeDesc","STRING")
						.column("eventOneTypeDesc","STRING")
						.column("eventTwoTypeDesc","STRING")
						.column("eventType","STRING")
						.column("extAttr","STRING")
						.column("fileHash","STRING")
						.column("fileName","STRING")
						.column("filePath","STRING")
						.column("idCard","STRING")
						.column("infectionFile","STRING")
						.column("insertTime","TIMESTAMP_LTZ(3)")
						.column("installNum","INTEGER")
						.column("intelId","STRING")
						.column("intelType","STRING")
						.column("logSubType","STRING")
						.column("logType","STRING")
						.column("mailFileSize","STRING")
						.column("mailNum","INTEGER")
						.column("mailProtocol","STRING")
						.column("mailRecipient","STRING")
						.column("mailSender","STRING")
						.column("mailType","STRING")
						.column("mailTypeDesc","STRING")
						.column("mainAccount","STRING")
						.column("malwareName","STRING")
						.column("malwareSubType","STRING")
						.column("malwareType","STRING")
						.column("mobile","STRING")
						.column("msg","STRING")
						.column("name","STRING")
						.column("operateContent","STRING")
						.column("operateType","STRING")
						.column("orgId","INTEGER")
						.column("orgPath","STRING")
						.column("password","STRING")
						.column("payload","STRING")
						.column("pcap","STRING")
						.column("peakByteFlow","INTEGER")
						.column("peakPackageFlow","INTEGER")
						.column("percentBaseLineFlow","STRING")
						.column("percentFlowTotal","STRING")
						.column("policy","STRING")
						.column("position","STRING")
						.column("protocol","STRING")
						.column("rawMsg","STRING")
						.column("registerNum","INTEGER")
						.column("responseCode","STRING")
						.column("result","INTEGER")
						.column("riskLevel","INTEGER")
						.column("riskLevelDesc","STRING")
						.column("sceneId","INTEGER")
						.column("source","INTEGER")
						.column("sourceEventThreeType","INTEGER")
						.column("sourceEventThreeTypeDesc","STRING")
						.column("sourceType","INTEGER")
						.column("srcAssetId","INTEGER")
						.column("srcAssetGroup","STRING")
						.column("srcAssetKey","INTEGER")
						.column("srcAssetName","STRING")
						.column("srcAssetPublic","INTEGER")
						.column("srcAssetStatus","INTEGER")
						.column("srcAssetSubType","STRING")
						.column("srcAssetType","STRING")
						.column("srcBizId","STRING")
						.column("srcCity","STRING")
						.column("srcCountry","STRING")
						.column("srcDomainName","STRING")
						.column("srcDomain2Name","STRING")
						.column("srcIntelDesc","STRING")
						.column("srcIntelId","STRING")
						.column("srcIntelType","INTEGER")
						.column("srcIp","STRING")
						.column("srcLatitude","STRING")
						.column("srcLongitude","STRING")
						.column("srcOrgId","STRING")
						.column("srcOrgName","STRING")
						.column("srcPort","STRING")
						.column("srcPost","STRING")
						.column("srcProvince","STRING")
						.column("srcSubDomainName","STRING")
						.column("subAccount","STRING")
						.column("tabelName","STRING")
						.column("tags","STRING")
						.column("tenantId","INTEGER")
						.column("terminalNum","INTEGER")
						.column("threatType","STRING")
						.column("totalByteFlow","INTEGER")
						.column("totalPackageFlow","INTEGER")
						.column("type","STRING")
						.column("upBaseLineFlow","INTEGER")
						.column("upFlowTotal","INTEGER")
						.column("url","STRING")
						.column("user","STRING")
						.column("userName","STRING")
						.column("userOrgName","STRING")
						.column("userType","STRING")
						.column("vulnId","STRING")
						.column("vulnInfo","STRING")
						.column("vulnLevel","STRING")
						.column("alertSignatureIdL","STRING")
						.column("vulnType","STRING")
						.column("alertInfo","STRING")
						.column("domain","STRING")
						.column("flag","STRING")
						.column("flow","INTEGER")
						.column("flowUp","INTEGER")
						.column("flowDown","INTEGER")
						.column("command","STRING")
						.primaryKey("snowId","createTime")
						.watermark("createTime", "SOURCE_WATERMARK()")

						.build());
	}
}
